package com.zhangpan.realtime.common.util;

import com.zhangpan.realtime.common.constant.Constant;

public class SQLUtil {

    public static String getKafkaDDLSource(String groupId,String topic) {
        return "with(" +
                " 'connector' = 'kafka'," +
                " 'properties.groupd.id'='" + groupId+"'," +
                " 'topic' = '" + topic + "',"+
                " 'properties.bootstrap.servers'='" +
                Constant.KAFKA_BROKERS + "'," +
                " 'scan.startup.mode'='latest-offset'," +
                " 'json.ignore-parse-errors'='true'," +
                " 'format' = 'json'" +
                ")";
    }

    public static String getKafkaDDLSink(String topic) {
        return "with(" +
                "  'connector' = 'kafka'," +
                "  'topic' = '" + topic + "'," +
                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "'," +
                "  'format' = 'json' " +
                ")";
    }

    public static String getUpsertKafkaDDL(String topic) {
        return "with(" +
                "  'connector' = 'upsert-kafka'," +
                "  'topic' = '" + topic + "'," +
                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "'," +
                "  'key.json.ignore-parse-errors' = 'true'," +
                "  'value.json.ignore-parse-errors' = 'true'," +
                "  'key.format' = 'json', " +
                "  'value.format' = 'json' " +
                ")";
    }
}
